package com.uuah.server.transport.provider.mina;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.mina.core.service.IoAcceptor;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.transport.socket.SocketAcceptor;
import org.apache.mina.transport.socket.SocketConnector;

import com.uuah.exception.UuahException;
import com.uuah.exception.lifecycle.InitialisationException;
import com.uuah.monitor.ExpiryMonitor;
import com.uuah.server.api.IUuahEvent;
import com.uuah.server.api.IUuahMessage;
import com.uuah.server.api.component.service.IService;
import com.uuah.server.api.endpoint.IUuahEndpoint;
import com.uuah.server.api.transport.provider.mina.IServerAcceptorFactory;
import com.uuah.server.transport.AbstractConnector;

/**
 *
 * TODO comment ...
 *
 * <p>
 * update record:updator，update time，update content and version
 * </p>
 *
 * @author <a href="jonny_quan@hotmail.com">jonny</a>
 * @date 2009 2009-3-27 下午02:19:26
 * @version 1.0.0
 */
public class MinaConnector extends AbstractConnector {

	private boolean sendOpenCloseEvents = false;
	private boolean sendIdleEvents = false;
	private boolean waitForSynchronousResponse = true;
	private int readerIdleTimeOut = MinaProperties.DEFAULT_SOCKET_TIMEOUT;
	private int writerIdleTimeOut = MinaProperties.DEFAULT_SOCKET_TIMEOUT;
	private int bothIdleTimeOut = MinaProperties.DEFAULT_SOCKET_TIMEOUT;
	private int sendBufferSize = MinaProperties.DEFAULT_BUFFER_SIZE;
	private int receiveBufferSize = MinaProperties.DEFAULT_BUFFER_SIZE;
	private int receiveBacklog = MinaProperties.DEFAULT_BACKLOG;
	private boolean reuseAddress = Boolean.TRUE;
	private boolean keepAlive = false;
	private int keepAliveTimeout = 0;
	private IServerAcceptorFactory serverAcceptorFactory;
	private ExpiryMonitor keepAliveMonitor;
	private ConcurrentHashMap<InetSocketAddress, ConcurrentHashMap<Number, IoSession>> servers;
	private AtomicLong maxId;


	public MinaConnector() {
		setServerAcceptorFactory(new MinaServerAcceptorFactory());
	}


	@Override
	public void doInitialise() throws InitialisationException {
		this.servers = new ConcurrentHashMap<InetSocketAddress, ConcurrentHashMap<Number, IoSession>>();
		this.maxId = new AtomicLong(1);
	}

	@Override
	public MinaMessageReceiver createReceiver(IService service,
			IUuahEndpoint endpoint) throws Exception {
		String hostname = endpoint.getEndpointURI().getHost();
		int port = endpoint.getEndpointURI().getPort();
		InetSocketAddress address = new InetSocketAddress(hostname, port);

		this.servers.putIfAbsent(address,
				new ConcurrentHashMap<Number, IoSession>());
		return (MinaMessageReceiver) getServiceDescriptor()
				.createMessageReceiver(this, service, endpoint);
	}

	public Number addSession(InetSocketAddress endpointAddress,
			IoSession session) {
		ConcurrentHashMap<Number, IoSession> sessionsMap = this.servers
				.get(endpointAddress);
		if (sessionsMap == null) {
			sessionsMap = new ConcurrentHashMap<Number, IoSession>();
		}
		Number id = maxId.incrementAndGet();
		sessionsMap.put(id, session);
		return id;
	}

	public Number addSession(URI endpointUri, IoSession session) {
		InetSocketAddress address = new InetSocketAddress(
				endpointUri.getHost(), endpointUri.getPort());
		return this.addSession(address, session);
	}

	public IoSession removeSession(URI endpointUri, Number sessionId) {
		InetSocketAddress address = new InetSocketAddress(
				endpointUri.getHost(), endpointUri.getPort());
		return removeSession(address, sessionId);
	}

	public IoSession removeSession(InetSocketAddress endpointAddress,
			Number sessionId) {
		ConcurrentHashMap<Number, IoSession> sessionsMap = this.servers
				.get(endpointAddress);
		return sessionsMap.remove(sessionId);
	}

	public IoSession getSession(URI endpointUri, Number sessionId) {
		InetSocketAddress address = new InetSocketAddress(
				endpointUri.getHost(), endpointUri.getPort());
		return getSession(address, sessionId);
	}

	public IoSession getSession(InetSocketAddress endpointAddress,
			Number sessionId) {
		ConcurrentHashMap<Number, IoSession> sessionsMap = this.servers
				.get(endpointAddress);
		return sessionsMap.get(sessionId);
	}

	public boolean hasServer(URI endpointUri) {
		InetSocketAddress address = new InetSocketAddress(
				endpointUri.getHost(), endpointUri.getPort());
		return this.servers.containsKey(address);
	}

	@Override
	protected void doConnect() throws Exception {

	}

	@Override
	protected void doDisconnect() throws Exception {

	}

	@Override
	protected void doDispose() {

	}

	@Override
	protected void doStart() throws UuahException {

	}

	@Override
	protected void doStop() throws UuahException {

	}

	public boolean isSendOpenCloseEvents() {
		return sendOpenCloseEvents;
	}

	public void setSendOpenCloseEvents(boolean sendOpenCloseEvents) {
		this.sendOpenCloseEvents = sendOpenCloseEvents;
	}

	public boolean isSendIdleEvents() {
		return sendIdleEvents;
	}

	public void setSendIdleEvents(boolean sendIdleEvents) {
		this.sendIdleEvents = sendIdleEvents;
	}

	public int getReaderIdleTimeOut() {
		return readerIdleTimeOut;
	}

	public void setReaderIdleTimeOut(int readerIdleTimeOut) {
		this.readerIdleTimeOut = readerIdleTimeOut;
	}

	public int getWriterIdleTimeOut() {
		return writerIdleTimeOut;
	}

	public void setWriterIdleTimeOut(int writerIdleTimeOut) {
		this.writerIdleTimeOut = writerIdleTimeOut;
	}

	public int getBothIdleTimeOut() {
		return bothIdleTimeOut;
	}

	public void setBothIdleTimeOut(int bothIdleTimeOut) {
		this.bothIdleTimeOut = bothIdleTimeOut;
	}

	public boolean isWaitForSynchronousResponse() {
		return waitForSynchronousResponse;
	}

	public void setWaitForSynchronousResponse(boolean waitForSynchronousResponse) {
		this.waitForSynchronousResponse = waitForSynchronousResponse;
	}

	public int getKeepAliveTimeout() {
		return keepAliveTimeout;
	}

	public void setKeepAliveTimeout(int keepAliveTimeout) {
		this.keepAliveTimeout = keepAliveTimeout;
	}

	public ExpiryMonitor getKeepAliveMonitor() {
		return keepAliveMonitor;
	}

	public void setKeepAliveMonitor(ExpiryMonitor keepAliveMonitor) {
		this.keepAliveMonitor = keepAliveMonitor;
	}

	public boolean isKeepAlive() {
		return keepAlive;
	}

	public void setKeepAlive(boolean keepAlive) {
		this.keepAlive = keepAlive;
	}

	public void setServerAcceptorFactory(
			IServerAcceptorFactory serverAcceptorFactory) {
		this.serverAcceptorFactory = serverAcceptorFactory;
	}

	public IServerAcceptorFactory getServerAcceptorFactory() {
		return serverAcceptorFactory;
	}

	public boolean isReuseAddress() {
		return reuseAddress;
	}

	public void setReuseAddress(boolean reuseAddress) {
		this.reuseAddress = reuseAddress;
	}

	public int getSendBufferSize() {
		return sendBufferSize;
	}

	public void setSendBufferSize(int sendBufferSize) {
		this.sendBufferSize = sendBufferSize;
	}

	public int getReceiveBufferSize() {
		return receiveBufferSize;
	}

	public void setReceiveBufferSize(int receiveBufferSize) {
		this.receiveBufferSize = receiveBufferSize;
	}

	public int getReceiveBacklog() {
		return receiveBacklog;
	}

	public void setReceiveBacklog(int receiveBacklog) {
		this.receiveBacklog = receiveBacklog;
	}

	public IoAcceptor getServerAcceptor(URI uri) {
		try {
			return getServerAcceptorFactory().createServerAcceptor(uri);
		} catch (IOException e) {
			logger.error(e.getMessage());
		}
		return null;
	}

	public void configureSocketConnector(SocketConnector socketConnector) {
		socketConnector.getSessionConfig().setKeepAlive(isKeepAlive());
		socketConnector.getSessionConfig().setKeepAlive(isKeepAlive());
		socketConnector.getSessionConfig().setReuseAddress(isReuseAddress());
	}

	public void configureSocketAcceptor(SocketAcceptor socketAcceptor) {
		socketAcceptor.getSessionConfig().setKeepAlive(isKeepAlive());
		socketAcceptor.getSessionConfig().setReuseAddress(isReuseAddress());
		socketAcceptor.setBacklog(getReceiveBacklog());
	}

	public String getProtocol() {
		return MinaProperties.MINA_PROTOCOL_PREFEX;
	}

	public void dispatch(IUuahMessage result, IUuahEvent event) {
		Number sessionId = (Number) event.getSession().getProperty(
				MinaProperties.MINA_SESSION_ID_PROPERTY);
		IoSession session = getSession(event.getEndpoint().getEndpointURI()
				.getUri(), sessionId);
		Object response = result.getPayload();
		if (response != null) {
			if (response instanceof List) {
				if (((List) response).size() > 0) {
					session.write(response);
				}
			} else {
				session.write(response);
			}
		}
	}
}
